package com.wpml.netty.nio.multi;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @Description 工作线程
 * @Date 2021-07-17
 * @Author wangpei
 **/
public class WorkerThread extends ThreadLocal<LinkedBlockingQueue<Channel>> implements Runnable {

    // 多路复用器
    Selector selector;
    /**
     * 工作线程组     每一个组持有各自的工作线程
     * 工作线程持有线程组的引用
     * */
    WorkerGroup workerGroup;

    LinkedBlockingQueue<Channel> lbq = get();

    @Override
    protected LinkedBlockingQueue<Channel> initialValue() {
        return new LinkedBlockingQueue<>();
    }

    public WorkerThread() {
        try {
            selector = Selector.open();
        } catch (IOException e) {
            System.out.println("工作线程初始化失败！！！");
            System.exit(1);
        }
    }


    @Override
    public void run() {

        while (true) {
            try {
                // 监听channel是否有读写事件
                int num = selector.select(); // 阻塞
                if (num > 0 ) {
                    // 得到事件channel
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> iter = selectionKeys.iterator();
                    SelectionKey key;
                    while (iter.hasNext()) {
                        key = iter.next();
                        iter.remove();
                        if (key.isAcceptable()) {
                            acceptHandler(key);
                        }
                        if (key.isReadable()) {
                            readHandler(key);
                        }
                        if (key.isWritable()) {
                            System.out.println("isWritable");
                        }
                    }
                }

                // 其他线程唤醒上面阻塞会执行下面代码
                // 从任务队列中获取任务处理
                if (!lbq.isEmpty()) {
                    Channel channel = lbq.take();
                    if (channel instanceof ServerSocketChannel) {
                        ServerSocketChannel server = (ServerSocketChannel) channel;
                        server.register(selector, SelectionKey.OP_ACCEPT); // 注册

                        System.out.println(Thread.currentThread().getName() + " server register listen port " + server.socket().getLocalPort());
                    }
                    if (channel instanceof SocketChannel) {
                        SocketChannel client = (SocketChannel) channel;
                        ByteBuffer buffer = ByteBuffer.allocateDirect(4096);
                        client.register(selector, SelectionKey.OP_READ, buffer);
                        System.out.println(Thread.currentThread().getName() + " client register listen port " + client.getRemoteAddress());
                    }
                }

            } catch (IOException | InterruptedException e) {
                e.printStackTrace();
            }
        }

    }

    private void readHandler(SelectionKey key) throws IOException {
        SocketChannel client = (SocketChannel) key.channel();
        ByteBuffer readBuf = (ByteBuffer) key.attachment();
        readBuf.clear();
        try {
            int read = client.read(readBuf);
            if (read > 0) {
                readBuf.flip();
                byte[] bytes = new byte[readBuf.remaining()];
                readBuf.get(bytes);
                String body = new String(bytes, Charset.defaultCharset());
                System.out.println(Thread.currentThread().getName() + "接收到的数据： " + body);

                // 组装响应数据
                String retStr = "服务端响应： " + body;

                doWrite(retStr, client);
            } else if (read < 0) {
                // 客户端断开了
                System.out.println("client : " + client.getRemoteAddress() + " 断开了......");
                key.cancel();
                key.channel().close();
            }
        } catch (IOException e) {
            e.printStackTrace();
            key.cancel();
            key.channel().close();
        }
    }

    private void doWrite(String retStr, SocketChannel client) throws IOException {
        byte[] bytes = retStr.getBytes(StandardCharsets.UTF_8);
        ByteBuffer writeBuf = ByteBuffer.allocate(bytes.length);
        writeBuf.put(bytes);
        writeBuf.flip();
        client.write(writeBuf);

        if (!writeBuf.hasRemaining()) {
            System.out.println("服务端响应完成");
        }
    }

    private void acceptHandler(SelectionKey key) {

        try {
            ServerSocketChannel server = (ServerSocketChannel) key.channel();
            // 等待客户端连接
            SocketChannel client = server.accept();
            client.configureBlocking(false);

            workerGroup.selectWorkThread(client);

        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * server 线程持有工作线程组引用
     * @param workerGroup
     */
    public void setWorkerGroup(WorkerGroup workerGroup) {
        this.workerGroup = workerGroup;
    }
}
